package jmind.core.rabbitmq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import jmind.base.lang.IProperties;
import jmind.base.util.DataUtil;

import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;

/**
 * Created by xieweibo on 2016/10/31.
 */
public  class RabbitmqProducer implements  Producer {


    private final Channel channel;
    private final String exchangeName ;

    public RabbitmqProducer(String name,IProperties p) throws IOException, TimeoutException {
        this(name,p,null,null,null,false);
    }

    /**
     对于已经建好的队列可以不调用 channel.exchangeDeclare 再次关联，可以传null
     持久化必须满足3个条件
     1： 消息投递模式选项设置为2  deliveryMode=2
     2：交换器持久化
     3： 队列持久化
     */
    public RabbitmqProducer(String name, IProperties p, String queueName, String exchangeName, ExchangeType type, boolean durable) throws IOException, TimeoutException {
        Connection connection = RabbitmqFactory.getFactory().getConnection(name, p);
        channel = connection.createChannel();
        this.exchangeName=exchangeName;
        //String message = "{\"service_order_id\":\"1111\",\"order_info\":{\"start_position\":\"中国技术交易大厦\",\"is_asap\":1,\"extension\":\"\",\"flag\":6291456,\"app_version\":\"iWeidao/7.0.0 D/881444\",\"city\":\"bj\",\"service_order_id\":6346758632293969658,\"dispatch_driver_ids\":\"\",\"source\":10000001,\"expect_start_latitude\":39.984051,\"end_position\":\"天通北苑二区\",\"expect_start_longitude\":116.307691,\"app_msg\":\"\",\"fixed_product_id\":0,\"contribution\":800,\"car_type_ids\":\"1\",\"expect_start_time\":1464776447,\"expect_end_longitude\":116.423779,\"time_length\":2810,\"create_time\":1464776447,\"flight_number\":\"\",\"product_type_id\":1,\"is_face_pay\":1,\"passenger_name\":\"小石头\",\"user_id\":3046655,\"corporate_id\":0,\"start_address\":\"北四环西路66号\",\"expect_end_latitude\":40.075948,\"corporate_dept_id\":0,\"end_address\":\"昌平区太平庄北街\"}}";
        if(!DataUtil.isEmpty(queueName))
        channel.queueDeclare(queueName, durable, false, false, null);
        if(!DataUtil.isEmpty(exchangeName))
        channel.exchangeDeclare(exchangeName, type.name(),durable);
        if (connection != null) {
            Runtime.getRuntime().addShutdownHook(new Thread() {
                @Override
                public void run() {
                    try {
                        channel.close();
                        connection.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }

    }

    public boolean produce(String routingKey,String message){
      return  produce(exchangeName,routingKey,message);

    }

    @Override
    public boolean produce(String exchange, String routingKey, String message) {
        try {
            channel.basicPublish(exchange,routingKey,null,message.getBytes());
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }
    // http://blog.csdn.net/u011001723/article/details/51887088
    // http://zhangyp.net/rabbitmq-delayqueue/
    @Override
    public boolean produce(String exchange, String routingKey, int delayTime, String message) {
        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
        // expiration 延时时间 ，deliveryMode 为2 表示持久化
        AMQP.BasicProperties properties = builder.expiration(String.valueOf(delayTime)).deliveryMode(2).build();
        try {
            String DELAY_QUEUE="DELAY_QUEUE_"+routingKey;
            // 实际要绑定的是queue。 这里通常routingKey 就是queue
            channel.queueBind(routingKey, "amq.direct", routingKey);
            HashMap<String, Object> arguments = new HashMap<String, Object>();
            arguments.put("x-dead-letter-exchange", "amq.direct");
            arguments.put("x-dead-letter-routing-key", routingKey);
            channel.queueDeclare(DELAY_QUEUE, true, false, false, arguments);
            channel.basicPublish("",DELAY_QUEUE,properties,message.getBytes());
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean produce(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body) {
        try {
            channel.basicPublish(exchange,routingKey,props,body);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }
}
